#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <cassert>
#include <sstream>
#include <list>

namespace mymq
{
    using MessagePtr = std::shared_ptr<MQMessage>;
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"
    class MessageMapper
    {
    public:
        MessageMapper(std::string& basedir, const std::string& qname)
        : _queuename(qname)
        {
            if (basedir.back() != '/') basedir += "/";

            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;

            assert(FileHelper::createDirectory(basedir));
            createMsgFile();
        }

        bool createMsgFile()
        {
            // 注意如果文件已存在，就不需要创建了
            // 否则，在进行 recovery 时，就可能会把已有的数据清空
            if (FileHelper(_datafile).exists() == true) return true;

            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建队列数据文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }

        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        bool insert(MessagePtr& mp)
        {
            return insert(_datafile, mp);
        }
        
        bool remove(MessagePtr& mp)
        {
            // 1.将消息的有效位置零
            mp->mutable_payload()->set_valid("0");

            // 2.序列化有效数据
            std::string body = mp->payload().SerializeAsString();
            if (body.size() != mp->length())
            {
                DLOG("删除队列数据失败：修改后的数据与原文件大小不一致");
                return false;
            }

            // 3.将有效数据写入到文件
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), mp->offset(), mp->length());
            if (ret == false)
            {
                DLOG("写入队列数据失败");
                return false;
            }
            // DLOG("确认消息后，删除持久化消息成功：%s", mp->payload().body().c_str());
            return true;
        }

        // 垃圾回收&返回有效数据
        std::list<MessagePtr> gc()
        {
            // 1.加载有效数据
            std::list<MessagePtr> result;
            bool ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效队列数据失败");
                return result;
            }

            // 2.将有效数据插入到临时文件
            FileHelper::createFile(_tmpfile);
            for (auto& msg : result)
            {
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    DLOG("有效队列数据插入到临时文件失败");
                    return result;
                }
            }

            // 3.删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败");
                return result;
            }

            // 4.重命名临时文件
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("重命名临时失败");
                return result;
            }

            // 5.返回有效数据
            return result;
        }

    private:
        bool load(std::list<MessagePtr>& result)
        {
            FileHelper helper(_datafile);
            size_t offset = 0, msg_size = 0;
            size_t fsize = helper.size();
            bool ret;
            while(offset < fsize)
            {
                // 读取消息长度
                ret = helper.read((char*)&msg_size, offset, sizeof(size_t));
                offset += sizeof(size_t);
                if (ret == false)
                {
                    DLOG("读取消息长度失败！");
                    return false;
                }
                // 读取消息
                std::string body(msg_size, '\0');
                ret = helper.read(&body[0], offset, msg_size);
                offset += msg_size;
                if (ret == false)
                {
                    DLOG("读取消息失败！");
                    return false;
                }

                // 构造结构化消息对象，反序化消息存储到对象
                MessagePtr msg = std::make_shared<MQMessage>();
                msg->mutable_payload()->ParseFromString(body);
                if (msg->payload().valid() == "0") continue; // 无效消息

                result.push_back(msg);
            }
            return true;
        }

        bool insert(const std::string& filename, MessagePtr& mp)
        {
            // 1.将消息的有效载荷进行序列化
            std::string body = mp->payload().SerializeAsString();

            // 2.获取文件大小，也就是写入数据时的偏移量
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();

            // 3.将数据写入文件指定位置
            // 先写入 4 字节的数据长度
            bool ret = helper.write((char*)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("写入队列数据长度失败！");
                return false;
            }
            // 写入队列数据
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), msg_size);
            if (ret == false)
            {
                DLOG("写入队列数据失败！");
                return false;
            }

            // 4.更新mp中的实际存储位置和大小
            mp->set_offset(fsize + sizeof(size_t));
            mp->set_length(msg_size);
            return true;
        }
    private:
        std::string _queuename;
        std::string _datafile;
        std::string _tmpfile;
    };

    // 以队列为单位进行消息管理
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string& basedir, const std::string& qname)
        : _qname(qname), _mapper(basedir, qname), _valid_count(0), _total_count(0)
        {}

        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc(); // 加载有效持久化消息
            for (auto& msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _total_count = _valid_count = _durable_msgs.size();
            return true;
        }

        bool insert(const BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            // 构造结构化消息对象
            MessagePtr mp = std::make_shared<MQMessage>();
            mp->mutable_payload()->set_body(body); // 消息主体部分
            // 消息属性
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(bp->id());
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                mp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                mp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                mp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                mp->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            // 将消息对象插入到内存管理的待推送队列
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs.push_back(mp);

            // 判断消息是否需要持久化
            if (mp->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                mp->mutable_payload()->set_valid("1");
                bool ret = _mapper.insert(mp);
                if (ret == false)
                {
                    DLOG("持久化存储消息 %s 失败！", body.c_str());
                    return false;
                }
                // 持久化成功
                _total_count++;
                _valid_count++;
                _durable_msgs.insert(std::make_pair(mp->payload().properties().id(), mp));
                // DLOG("插入持久化消息成功，持久化消息数量：%ld", _durable_msgs.size());
            }
            return true;
        }

        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0) return MessagePtr();

            // 从 待推送消息队列 中取出一条消息
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 放入到待确认消息hash
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        // 删除消息，也就是确认消息 每次删除持久化消息后，要进行垃圾回收
        bool remove(const std::string& msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            // 查看待确认消息是否存在
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                DLOG("没有找到要删除的消息：%s", msg_id.c_str());
                return true;
            }

            // 要删除的消息存在
            // 检查持久化标志，决定是否要删除持久化数据
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count--;
                gc();
            }
            // 删除内存中的消息
            _waitack_msgs.erase(msg_id);
            // DLOG("确认消息后，删除消息的内存管理成功：%s", it->second->payload().body().c_str());
            return true;
        }

        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _waitack_msgs.clear();
            _durable_msgs.clear();
            _valid_count = _total_count = 0;
        }
    private:
        bool GCCheck()
        {
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                return true;
            return false;
        }

        void gc()
        {
            if (GCCheck() == false) return;
            // 获取垃圾回收后，有效消息链表
            std::list<MessagePtr> msgs = _mapper.gc();

            // 更新持久化消息的存储信息
            for (auto& msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    // 不应该存在的情况
                    DLOG("垃圾回收后，有一条持久化消息没有在内存中进行管理！");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                // 更新存储信息到持久化消息hash
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }

            // 更新持久化消息数量
            _valid_count = _total_count = msgs.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        MessageMapper _mapper; // 消息数据持久化句柄
        size_t _valid_count; // 有效持久化消息数量
        size_t _total_count; // 总体持久化消息数量
        std::list<MessagePtr> _msgs; // 待推送消息队列
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息hash
    };

    // 对外提供的队列消息管理
    class MessgeManager
    {
    public:
        using ptr = std::shared_ptr<MessgeManager>;
        MessgeManager(const std::string& basedir)
        : _basedir(basedir)
        {}

        void initQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    // 队列消息管理句柄已存在
                    return;
                }
                // 创建句柄
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
             // 因为recovery里面的操作可能较为复杂，所以放在锁外面。不然可能导致锁冲突，_queue_msgs一直锁着，其他接口只能等待
            qmp->recovery();
        }

        void destroyQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    // 队列消息管理句柄不存在
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(qname);
            }
            qmp->clear(); // 同理，防止锁冲突
        }

        bool insert(const std::string& qname, BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列 %s 新增消息失败：找不到队列消息管理句柄", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }

        void ack(const std::string& qname, const std::string& msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认队列 %s 消息 %s 失败：找不到队列消息管理句柄", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }

        MessagePtr front(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 队首消息 %s 失败：找不到队列消息管理句柄", qname.c_str(), qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }

        size_t getable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 待推送消息数量失败：找不到队列消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }

        size_t waitack_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 待确认消息数量失败：找不到队列消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

        size_t durable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 有效持久化消息数量失败：找不到队列消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }

        size_t total_count(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 总持久化消息数量失败：找不到队列消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto& qmsg : _queue_msgs)
                qmsg.second->clear();
        }
    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}
#endif